package com.jxufe.weather;

import java.io.IOException;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class WeaReducer extends Reducer<Text, WeaBean, Text, WeaStatBean> {

	@Override
	protected void reduce(Text key, Iterable<WeaBean> values, Context context)
			throws IOException, InterruptedException {
		int cityid = 0;
		double tem = 0;

		double maxLowTem = 0, minLowTem = 100, averLowTem = 0;

		double maxHighTem = 0, minHighTem = 100, averHighTem = 0;

		double sumHigh = 0, sumLow = 0;

		int count = 0;

		for (WeaBean value : values) {
			if (count == 0) {
				cityid = value.getCityid();
				tem = value.getTem();
//				throw new RuntimeException(key + ":"+value.toString());
				// 问题可能是由于bean中写入字符串方法导致
//				WeatherBean [cityid=1010101006åäº¬32â22â26â, week=null, city=null, tem1=null, tem2=null, tem=null]
			}
			double highTem = value.getTem1();// 高温

			double lowTem = value.getTem2();// 低温

			sumHigh += highTem;
			if (highTem > maxHighTem)
				maxHighTem = highTem;
			if (highTem < minHighTem)
				minHighTem = highTem;

			sumLow += lowTem;
			if (lowTem > maxLowTem)
				maxLowTem = lowTem;
			if (lowTem < minLowTem)
				minLowTem = lowTem;
			count++;
		}
		averHighTem = sumHigh / count;
		averHighTem = Double.parseDouble(String.format("%.1f", averHighTem));

		averLowTem = sumLow / count;
		averLowTem = Double.parseDouble(String.format("%.1f", averLowTem));
		WeaStatBean statBean = new WeaStatBean(cityid, tem, maxLowTem, minLowTem, averLowTem, maxHighTem, minHighTem,
				averHighTem);
		context.write(key, statBean);
		System.out.println("reduce end......");
	}
//	private String TemSubstring(String temStr) {
//		int index = temStr.indexOf('℃');
//		return temStr.substring(0,index);
//	}

}
